package com.hudson.broker.handler.process;

import com.hudson.broker.handler.Process;
import com.hudson.broker.consumer.PullRequest;
import com.hudson.broker.consumer.WaitRequest;
import com.hudson.broker.monitor.BizMonitor;
import com.hudson.codec.domain.MessageExt;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.domain.request.PushMessageReq;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.json.JSONUtil;
import com.hudson.core.util.CollectionUtil;
import com.hudson.store.commitlog.CommitLogManger;
import com.hudson.store.consumer.ConsumerOffset;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

public class ConsumerProcess implements Process {

    private static final Logger logger = LoggerFactory.getLogger(ConsumerProcess.class);


    private static final ConcurrentHashMap<String, Integer> consumerOffsetMap = new ConcurrentHashMap<>();


    @Override
    public RemotingCommand process(ChannelHandlerContext context, RemotingCommand remotingCommand) {
        logger.debug("Serve 收到消息:{}", remotingCommand);
        PushMessageReq pushMessageReq = JSONUtil.fromJson(new String(remotingCommand.getBody(), StandardCharsets.UTF_8), PushMessageReq.class);
        logger.debug("{}", pushMessageReq);

        // 返回请求再更新消费offset
        String key = pushMessageReq.getConsumerGroup() + pushMessageReq.getTopic() + pushMessageReq.getQueueId();

        int offset;
        if (consumerOffsetMap.containsKey(key)) {
            ConsumerOffset.putOffset(pushMessageReq.getConsumerGroup(), pushMessageReq.getTopic(), pushMessageReq.getQueueId(), consumerOffsetMap.get(key));
            offset = consumerOffsetMap.get(key);
        } else {
            offset = ConsumerOffset.getOffset(pushMessageReq.getConsumerGroup(), pushMessageReq.getTopic(), pushMessageReq.getQueueId());
        }
        List<MessageExt> realMessageList = CommitLogManger.getListMessageExtByTopic(pushMessageReq.getTopic(), pushMessageReq.getQueueId(), offset, pushMessageReq.getSize());
        consumerOffsetMap.put(key, offset + realMessageList.size());


        if (CollectionUtil.isEmpty(realMessageList)) {

            WaitRequest.putPullRequest(pushMessageReq.getTopic(), pushMessageReq.getQueueId(), new PullRequest(context, pushMessageReq.getConsumerGroup(), pushMessageReq.getTopic(), pushMessageReq.getQueueId(), System.currentTimeMillis()));
            return null;
        } else {

            BizMonitor.CONSUME_SPEED.getAndIncrement();

            return RemotingCommand.builder()
                    .type(MessageTypeEnum.PUSH_MESSAGE_RESP.getType())
                    .opaque(remotingCommand.getOpaque())
                    .body(MessageExt.list2Bytes(realMessageList))
                    .build();
        }
    }
}
